//   Copyright 2012,2013 Vaughn Vernon
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//       http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
package com.saasovation.common.port.adapter.persistence.eventsourcing.mysql;

import com.saasovation.common.domain.model.DomainEvent;
import com.saasovation.common.event.EventSerializer;
import com.saasovation.common.event.sourcing.*;
import com.saasovation.common.port.adapter.persistence.eventsourcing.DefaultEventStream;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static java.lang.Class.forName;

public class MySQLEventStore implements EventStore {

    private EventNotifiable eventNotifiable;
    private EventSerializer serializer;
    private final NamedParameterJdbcTemplate jdbcTemplate;

    public MySQLEventStore(NamedParameterJdbcTemplate jdbcTemplate) {
        super();
        this.jdbcTemplate = jdbcTemplate;
        this.setSerializer(EventSerializer.instance());
    }

    @Override
    public void appendWith(EventStreamId aStartingIdentity, List<DomainEvent> anEvents) {
        // tbl_es_event_store must have a composite primary key
        // consisting of {stream_name}:{streamVersion} so that
        // appending a stale version will fail the pk constraint
        int index = 0;
        for (DomainEvent event : anEvents) {
            try {
                this.appendEventStore(aStartingIdentity, index++, event);
            } catch (DataAccessException e) {
                throw new EventStoreAppendException(e.getMessage(), e);
            }
        }
        this.notifyDispatchableEvents();
    }

    @Override
    public void close() {
        // no-op
    }

    @Override
    public List<DispatchableDomainEvent> eventsSince(long aLastReceivedEvent) {
        String sql = "select event_id, event_body, event_type from tbl_es_event_store "
                + "where event_id > :eventId order by event_id ";
        Map<String, Object> params = new HashMap<>();
        params.put("eventId", aLastReceivedEvent);
        return this.jdbcTemplate.query(sql, params, dispatchableDomainEventRowMapper());
    }

    @Override
    public EventStream eventStreamSince(EventStreamId anIdentity) {
        String sql = "select stream_version, event_type, event_body from tbl_es_event_store "
                + "where stream_name = :streamName and stream_version >= :streamVersion "
                + "order by stream_version";
        Map<String, Object> params = new HashMap<>();
        params.put("streamName", anIdentity.streamName());
        params.put("streamVersion", anIdentity.streamVersion());
        DefaultEventStream eventStream = queryEventStream(sql, params);
        if (eventStream.version() == 0) {
            throw new EventStoreException("There is no such event stream: "
                    + anIdentity.streamName()
                    + " : "
                    + anIdentity.streamVersion());
        }
        return eventStream;
    }

    @Override
    public EventStream fullEventStreamFor(EventStreamId anIdentity) {
        String sql = "select stream_version, event_type, event_body " +
                "from tbl_es_event_store where stream_name = :streamName " +
                "order by stream_version";
        Map<String, Object> params = new HashMap<>();
        params.put("streamName", anIdentity.streamName());
        return queryEventStream(sql, params);
    }

    public long greatestEventId() {
        String sql = "select max(event_id) from tbl_es_event_store ";
        Long lastEventId = null;
        try {
            lastEventId = this.jdbcTemplate.queryForObject(sql, new HashMap<>(), Long.class);
        } catch (EmptyResultDataAccessException e) {
            // ignore
        }
        return lastEventId == null ? 0 : lastEventId;
    }

    @Override
    public void purge() {
        String sql = "delete from tbl_es_event_store ";
        this.jdbcTemplate.update(sql, new HashMap<>());
    }

    @Override
    public void registerEventNotifiable(EventNotifiable anEventNotifiable) {
        this.eventNotifiable = anEventNotifiable;
    }

    private void appendEventStore(EventStreamId anIdentity,
                                  int anIndex,
                                  DomainEvent aDomainEvent) {
        String sql = "insert into tbl_es_event_store(event_body, event_type, stream_name, stream_version) " +
                "values(:eventBody, :eventType, :streamName, :streamVersion)";
        Map<String, Object> params = new HashMap<>();
        params.put("eventBody", this.serializer().serialize(aDomainEvent));
        params.put("eventType", aDomainEvent.getClass().getName());
        params.put("streamName", anIdentity.streamName());
        params.put("streamVersion", anIdentity.streamVersion() + anIndex);
        this.jdbcTemplate.update(sql, params);
    }

    private DefaultEventStream queryEventStream(String sql, Map<String, Object> params) {
        List<DomainEvent> events = new ArrayList<>();
        VersionHolder version = new VersionHolder();
        this.jdbcTemplate.query(sql, params, rs -> {
            version.steamVersion = rs.getInt("stream_version");
            String eventClassName = rs.getString("event_type");
            String eventBody = rs.getString("event_body");
            DomainEvent domainEvent = deserializeEvent(eventClassName, eventBody);
            events.add(domainEvent);
        });
        return new DefaultEventStream(events, version.steamVersion);
    }

    private RowMapper<DispatchableDomainEvent> dispatchableDomainEventRowMapper() {
        return (rs, rowNum) -> {
            long eventId = rs.getLong("event_id");
            String eventClassName = rs.getString("event_type");
            String eventBody = rs.getString("event_body");
            DomainEvent domainEvent = deserializeEvent(eventClassName, eventBody);
            return new DispatchableDomainEvent(eventId, domainEvent);
        };
    }

    @SuppressWarnings("unchecked")
    private DomainEvent deserializeEvent(String eventClassName, String eventBody) {
        Class<DomainEvent> eventClass;
        try {
            eventClass = (Class<DomainEvent>) forName(eventClassName);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
        return serializer().deserialize(eventBody, eventClass);
    }

    private static class VersionHolder {
        int steamVersion;
    }

    private EventNotifiable eventNotifiable() {
        return this.eventNotifiable;
    }

    private void notifyDispatchableEvents() {
        EventNotifiable eventNotifiable = this.eventNotifiable();
        if (eventNotifiable != null) {
            this.eventNotifiable().notifyDispatchableEvents();
        }
    }

    private EventSerializer serializer() {
        return this.serializer;
    }

    private void setSerializer(EventSerializer aSerializer) {
        this.serializer = aSerializer;
    }
}
